package com.sky.rocket.consumer.config;

import com.sky.rocket.consumer.listener.RocketMessageListener;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RocketMQConfig {
    private String namesrvAddr = "192.168.31.78:9876";
    private String groupName = "sky";
    private int consumeThreadMin = 20;
    private int consumeThreadMax = 60;
    private String topics = "rocketKey";

    @Autowired
    private RocketMessageListener rocketMessageListener;

    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer() throws Exception{
        if (StringUtils.isEmpty(groupName)){
            throw new Exception("groupName is null !!!");
        }
        if (StringUtils.isEmpty(namesrvAddr)){
            throw new Exception("namesrvAddr is null !!!");
        }
        if(StringUtils.isEmpty(topics)){
            throw new Exception("topics is null !!!");
        }
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(rocketMessageListener);
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 设置一次消费消息的条数，默认为1条
         */
        consumer.setConsumeMessageBatchMaxSize(1);
        try {
            /**
             * 设置该消费者订阅的主题 和 tag * 表示该主题下所有tag
             */
            consumer.subscribe(topics, "*");
            consumer.start();
        } catch (Exception e) {
            throw new Exception(e);
        }
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer2() throws Exception{
        if (StringUtils.isEmpty(groupName)){
            throw new Exception("groupName is null !!!");
        }
        if (StringUtils.isEmpty(namesrvAddr)){
            throw new Exception("namesrvAddr is null !!!");
        }
        if(StringUtils.isEmpty(topics)){
            throw new Exception("topics is null !!!");
        }
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("skyTopic");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(rocketMessageListener);
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         * 设置消费模型，集群还是广播，默认为集群
         */
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 设置一次消费消息的条数，默认为1条
         */
        consumer.setConsumeMessageBatchMaxSize(1);
        try {
            /**
             * 设置该消费者订阅的主题 和 tag * 表示该主题下所有tag
             */
            consumer.subscribe("skyTopic", "*");
            consumer.start();
        } catch (Exception e) {
            throw new Exception(e);
        }
        return consumer;
    }
}
